/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.cst.lls.se;

import cn.ac.iie.di.cst.lls.conf.ConfigurationException;
import cn.ac.iie.di.cst.lls.common.RuntimeEnv;
import static cn.ac.iie.di.cst.lls.common.RuntimeEnv.SendMessageSize;
import cn.ac.iie.di.cst.lls.util.DataInfo;
import cn.ac.iie.di.cst.lls.util.DecodeMessage;
import cn.ac.iie.di.cst.lls.util.KillHandler;
import cn.ac.iie.di.cst.lls.util.RocketMQProducer;
import cn.ac.iie.di.cst.lls.worker.RocketdxConsumer;
import cn.ac.iie.di.cst.lls.worker.SendMsgToMq;
import static cn.ac.iie.di.cst.lls.worker.RocketdxConsumer.consumeData;
import com.alibaba.rocketmq.client.exception.MQClientException;
import java.util.HashMap;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.logging.Level;
import org.apache.log4j.Logger;
import org.apache.log4j.PropertyConfigurator;

/**
 */
public class CstdxStartup {

    public static ExecutorService sendMessagePool = Executors.newFixedThreadPool(30);
    public static ExecutorService decodePool = Executors.newFixedThreadPool(30);
    public static ArrayBlockingQueue<DataInfo> ab = new ArrayBlockingQueue<DataInfo>(3000);
    private static org.apache.log4j.Logger logger = Logger.getLogger(CstdxStartup.class.getName());
    public static String confPath;

    public static void main(String[] args) throws MQClientException, ConfigurationException {
//        CstdxStartup.confPath = args[0];
//        CstdxStartup.confPath = "conf";
        PropertyConfigurator.configure("log4j.properties");
        long starttime = System.currentTimeMillis();
        logger.info("starttime" + starttime);
        if (!RuntimeEnv.initialize()) {
            logger.error("RuntimeEnv not insert");
            System.exit(1);
        }
        String sss = (String) RuntimeEnv.getParam("mqList");
        String[] mqlist = sss.split("[|]");
        String pnamesrv = (String) RuntimeEnv.getParam("producerNamesrv");
        try {
            RocketMQProducer.init("ggcc_produce", pnamesrv);
        } catch (Exception E) {
            logger.error("ggcc_produceException" + E);
            System.exit(1);
        }
        consumeData((String) RuntimeEnv.getParam("consumerNamesrv"), sss);

        for (int p = 0; p < mqlist.length; p++) {
            SendMsgToMq smtq = new SendMsgToMq(mqlist[p]);
            DecodeMessage dMsg = new DecodeMessage();
            logger.info(" mq init successfully " + mqlist[p]);
            for (int i = 0; i < SendMessageSize; i++) {
                decodePool.submit(dMsg);
            }
            try {
                Thread.sleep(1000);
            } catch (InterruptedException ex) {
                ex.printStackTrace();
            }
            for (int i = 0; i < SendMessageSize; i++) {
                sendMessagePool.submit(smtq);
            }
        }

        KillHandler killhandle = new KillHandler();
        killhandle.registerSignal("TERM");
        logger.info("init the kill handler done ");
//        while(true){
//            try {
//                logger.info("at "+System.currentTimeMillis()+" form "+(String) RuntimeEnv.getParam("consumerNamesrv")+" consume  "+RocketdxConsumer.a.get()+"  messages");
//                Thread.sleep(100000);
//            } catch (InterruptedException ex) {
//                
//            }
//        }
    }
}
